package com.galeno.day11;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.Properties;


/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2920:33
 */
public class KafkaToKafka {
    public static void main(String[] args) throws Exception {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",11111);
        StreamExecutionEnvironment e = StreamExecutionEnvironment.getExecutionEnvironment();
        //开启checkpoint，那就把kafka的偏移量保存到状态中了，checkpoint时会将状态持久化到statebackend中
        e.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
        e.setStateBackend(new FsStateBackend("file:///D:/a"));
        String kafkaNode="galeno01:9092,galeno02:9092,galeno03:9092";
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers",kafkaNode);
        properties.setProperty("group.id","galeno");
        properties.setProperty("auto.offset.reset","latest");
        //如果要开启事务,需要将客户端的事务超时时间啊小于broker的事务超时时间,broker默认值为15分钟
        properties.setProperty("transaction.timeout.ms","600000");
        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>(
                "liu", new SimpleStringSchema(), properties
        );
        /**
         * do not  write to Kafka's special topic
         */
        flinkKafkaConsumer.setCommitOffsetsOnCheckpoints(false);

        DataStreamSource<String> lines = e.addSource(flinkKafkaConsumer);
        /**
         * use SocketSource default num,Use simulation data
         */
        DataStreamSource<String> simulationData = e.socketTextStream("galeno", 9999);
        simulationData.map(x->{
            if(x.contains("omg")){
               throw  new RuntimeException("err");
            }
            return x;
        }).union(lines);

        SingleOutputStreamOperator<String> filter = simulationData.filter(x -> !x.contains("app"));

        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
                "test", new MyKafkaSerializationSchema("test")
                , properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
        filter.addSink(kafkaProducer);
        e.execute();


    }
}
